package net.bw.realtime.tmall.dws.funcion1;

import net.bw.realtime.tmall.dwd.bean.TransactionEventBean;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import net.bw.realtime.tmall.common.utils.JdbcUtil;

import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * @Author: liuyawei
 * @Date: 2025/5/27
 */

//todo c 热门商品排行
public class ToHotSkuWindowFunction {
	public static void toHotSku(DataStream<TransactionEventBean> timeStream) {

		SingleOutputStreamOperator<String> hotsku = timeStream.filter(x -> "pay_success".equals(x.getActionType()))
				.keyBy(x -> x.getSkuId())
				.window(TumblingEventTimeWindows.of(Time.seconds(10)))
				.aggregate(new AggregateFunction<TransactionEventBean, Tuple3<Double,Long, String>, Tuple3<Double,Long, String>>() {

					@Override
					public Tuple3<Double,Long, String> createAccumulator() {
						return Tuple3.of(0.0,0L, "");
					}

					@Override
					public Tuple3<Double,Long, String> add(TransactionEventBean TransactionEventBean, Tuple3<Double,Long, String> doubleStringTuple3) {
						Double amount = TransactionEventBean.getQuantity() * TransactionEventBean.getPrice();
						return Tuple3.of(amount + doubleStringTuple3.f0,TransactionEventBean.getQuantity()  + doubleStringTuple3.f1, TransactionEventBean.getSku_name());
					}

					@Override
					public Tuple3<Double,Long, String> getResult(Tuple3<Double,Long, String> doubleStringTuple3) {

						return doubleStringTuple3;
					}

					@Override
					public Tuple3<Double,Long, String> merge(Tuple3<Double,Long, String> d1, Tuple3<Double,Long, String> acc1) {
						return Tuple3.of(d1.f0 + acc1.f0,d1.f1+acc1.f1, acc1.f2);
					}
				}, new ProcessWindowFunction<Tuple3<Double,Long, String>, String, String, TimeWindow>() {
					private final SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

					@Override
					public void process(String s, ProcessWindowFunction<Tuple3<Double,Long, String>, String, String, TimeWindow>.Context context, Iterable<Tuple3<Double,Long, String>> iterable, Collector<String> collector) throws Exception {
						Tuple3<Double,Long, String> next = iterable.iterator().next();
						String window_start = sdf.format(new Date(context.window().getStart()));
						String window_end = sdf.format(new Date(context.window().getEnd()));

						collector.collect(window_start + "," + window_end + "," + next.f2  + "," + next.f1 + "," + String.format("%.8E", next.f0) + "," + System.currentTimeMillis());
					}
				} )
				.name("Hot-sku-aggregation")
				.uid("Hot-sku-aggregation");

		//hotsku.print();

		JdbcUtil.sinkToClickhouseUpsert(hotsku,"INSERT INTO jtp_tmall_report.dws_sku_count_window_report (\n" +
				"     window_start_time, window_end_time, sku_name, sku_count, sku_total_amount, ts\n" +
				") VALUES (\n" +
				"    ?, ?, ?, ?, ?, ?\n" +
				")"
		);
	}


}
